package com.example.mq.mqclient;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.common.network.BasicReturns;
import com.example.mq.common.network.Request;
import com.example.mq.common.network.Response;
import com.example.mq.common.network.expansion.SubscribeReturns;
import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.net.Socket;
import java.net.SocketException;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 这个类表示一次完整的tcp连接
 */
@Slf4j
public class Connection {
    private Socket socket;
    //  一次连接包含多个channel
    private ConcurrentHashMap<String,Channel> channelMap = new ConcurrentHashMap<>();
    //  引入线程池，后续可能要执行回调
    private ExecutorService callback = Executors.newCachedThreadPool();
    //  引入流对象
    private InputStream inputStream;
    private OutputStream outputStream;
    private DataInputStream dataInputStream;
    private DataOutputStream dataOutputStream;

    public Connection(String host,int port) throws IOException {
        socket = new Socket(host,port);
        inputStream = socket.getInputStream();
        outputStream = socket.getOutputStream();
        dataInputStream = new DataInputStream(inputStream);
        dataOutputStream = new DataOutputStream(outputStream);

        //  引入扫描线程，获取服务器的响应
        Thread t = new Thread(() -> {
            while(!socket.isClosed()) {
                try {
                    Response response = readResponse();
                    dispatchResponse(response);

                } catch (SocketException e) {
                    //  正常运行
                } catch (IOException | ClassNotFoundException | MqException e) {
                    e.printStackTrace();
                }

            }
        });
        t.start();

        log.info("客户端已上线. " + socket);
    }

    public void close() throws IOException {
        channelMap.clear();
        inputStream.close();
        outputStream.close();
        socket.close();
    }

    //  创建一个 channel
    public Channel createChannel() throws MqException, IOException {
        String channelId = "C-" + UUID.randomUUID();
        Channel channel = new Channel(channelId,this);

        //  注意：要先保存channel，再判断是否成功
        //  因为创建 channel 时处理服务器发送过来的响应时，需要使用channel
        //  如果没有保存那么就创建不成功，不成功又不能保存，就会陷入先有鸡还有是先有蛋的问题
        channelMap.put(channelId,channel);

        //  创建 channel 时，要告知服务器
        boolean ok = channel.createChannel();
        if(!ok) {
            channelMap.remove(channelId);
            throw new MqException("创建 channel 失败! channelId=" + channelId);
        }

        log.info("创建 channel 成功. ");
        return channel;
    }

    private void dispatchResponse(Response response) throws IOException, ClassNotFoundException, MqException {
        //  1. 判断响应类型
        int type = response.getType();
        if(type == 0xc) {
            //  服务器主动推送的消息（生产者生产的消息）
            //  2. 反序列化响应
            SubscribeReturns subscribeReturns = (SubscribeReturns) BinaryTool.fromBytes(response.getPayload());
            Channel channel = channelMap.get(subscribeReturns.getChannelId());
            if(channel == null) {
                throw new MqException("消息对应的 channel 在客户端已经不存在! ");
            }
            //  3.执行回调
            callback.submit(() -> {
                try {
                    channel.getConsume().handleDelivery(subscribeReturns.getConsumerTag(),subscribeReturns.getBasicProperties(),subscribeReturns.getBody());
                } catch (MqException | IOException e) {
                    e.printStackTrace();
                }
            });
        }
        else {
            //  服务器完成客户端的请求后的响应
            //  2. 反序列化响应
            BasicReturns basicReturns = (BasicReturns) BinaryTool.fromBytes(response.getPayload());
            Channel channel = channelMap.get(basicReturns.getChannelId());
            if(channel == null) {
                throw new MqException("消息对应的 channel 在客户端已经不存在! ");
            }

            //  2. 把收到的响应存入 map
            channel.putReturn(basicReturns);
        }

    }

    private Response readResponse() throws IOException {
        int type = dataInputStream.readInt();
        int length = dataInputStream.readInt();
        byte[] payload = new byte[length];
        int read = dataInputStream.read(payload);
        if(read != length) {
            throw new IOException("解析的响应不完整! ");
        }

        Response response = new Response();
        response.setType(type);
        response.setLength(length);
        response.setPayload(payload);
        log.info("收到响应. response=" + response);
        return response;
    }

    protected void writeRequest(Request request) throws IOException {
        dataOutputStream.writeInt(request.getType());
        dataOutputStream.writeInt(request.getLength());
        dataOutputStream.write(request.getPayload());
        dataOutputStream.flush();
        log.info("发送请求. type=" + request.getType());
    }
}
